package com.hou.mqacting.service.impl;

import com.hou.mqacting.pojo.User;
import com.hou.mqacting.service.MessageSendService;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.ConfirmListener;
import com.rabbitmq.client.Envelope;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.DefaultMessagePropertiesConverter;
import org.springframework.amqp.rabbit.support.MessagePropertiesConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

/**
 * 消息接受成功回调需要实现ConfirmCallback接口
 */
@Service
public class MessageSendServiceImpl implements RabbitTemplate.ConfirmCallback,MessageSendService {

    @Value("${rabbit.queue.string}")
    private String strQueueName;

    @Value("${rabbit.queue.object}")
    private String objQueueName;

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Qualifier("testQueue")

    @Override
    public void sendString(String message) {
        //设置回调接口
        rabbitTemplate.setConfirmCallback(this);
        //发送消息到指定队列
        rabbitTemplate.convertAndSend(strQueueName,message);
    }
    @Override
    public void sendObject(User user) {
        //设置回调接口
        rabbitTemplate.setConfirmCallback(this);
        rabbitTemplate.convertAndSend(objQueueName,user);
    }
    //回调确认方法
    @Override
    public void confirm(CorrelationData correlationData, boolean b, String s) {
        if(b){
            System.out.println("消息发送成功");
        }else{
            System.out.println("消息消费失败:"+s);
        }
    }

    @Override
    public void sendTest() {
        /**
         * Mandatory为true时,消息通过交换器无法匹配到队列会返回给生产者
         *          为false时,匹配不到会直接被丢弃
         */
//        rabbitTemplate.setMandatory(true);
//        //添加监听器获取返送失败的消息
//        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
//            @Override
//            public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
//                System.out.println("replyCode:"+replyCode);
//                System.out.println("replyText:"+replyText);
//                System.out.println("匹配队列失败,返回消息:" + message.toString());
//            }
//        });
        // 向指定交换器发送消息,需要key
        rabbitTemplate.convertAndSend("exchange","routeKey","测试消息内容");
    }


    @Override
    public void sendBackExchange() {
        // 向指定交换器发送消息,需要key
        //指定备份交互器后可直接发送消息即可,匹配不到的消息会存储到备份队列里
        rabbitTemplate.convertAndSend("exchange","route","测试备份交互器");
    }


    //发送过期消息
    @Override
    public void sendTTLMessage() {
        // MessagePostProcessor: 可设置消息属性,设置消息过期时间
        rabbitTemplate.convertAndSend(strQueueName, (Object) "测试过期消息", new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                MessageProperties messageProperties = message.getMessageProperties();
                //设置过期时间
                messageProperties.setExpiration("10000");//10秒
                //设置消息持久化
                messageProperties.setDeliveryMode(MessageDeliveryMode.PERSISTENT);
                return message;
            }
        });
    }

    @Override
    public void sendPri1(Integer pri) {
        //发送具有优先级的消息,最大不能超过队列的设置优先级\
        rabbitTemplate.convertAndSend("prechange","pre1Key","队列preQueue1--第"+pri+"条消息", new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                MessageProperties messageProperties = message.getMessageProperties();
                messageProperties.setPriority(pri);
                return message;
            }
        });
    }

    @Override
    public void sendPri2(Integer pri) {
        //发送具有优先级的消息,最大不能超过队列的设置优先级
        rabbitTemplate.convertAndSend("prechange","pre2Key","队列preQueue2--第"+pri+"条消息", new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                MessageProperties messageProperties = message.getMessageProperties();
                messageProperties.setPriority(pri);
                return message;
            }
        });
    }

    //测试生产者确认模式
    @Override
    public void sendPCom() {
        //设置消息异步回调,当消息成功发送到mq时会回调该方法
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            //消息成功异步回调
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                if(ack){
                    System.out.println("消息发送成功:"+correlationData.getId());
                } else{
                    System.out.println("消息发送失败:"+cause);
                }
            }
        });
        //设置消息发送失败异步退回
        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            @Override
            public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
                System.out.println("发送失败的消息:"+message.toString());
            }
        });

    }


    public void sendRPC(Integer pri) {
        //发送具有优先级的消息,最大不能超过队列的设置优先级
        rabbitTemplate.convertAndSend("prechange","pre2Key","队列preQueue2--第"+pri+"条消息", new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                MessageProperties messageProperties = message.getMessageProperties();
                messageProperties.setPriority(pri);
                messageProperties.setReplyTo("rpc_queue");//设置回调队列
                messageProperties.setCorrelationId("123456");//唯一标识,匹配
                return message;
            }
        });
    }


}
